package calao.storage.bdb;

import static calao.util.Logger.*;

import java.util.Vector;

import calao.storage.Search;
import calao.storage.Table;
import calao.storage.Record;
import calao.storage.Field;
import calao.storage.Index;
import calao.storage.FieldCondition;
import calao.storage.EqualCondition;
import calao.storage.RangeCondition;
import calao.util.Cursor;

import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.bind.tuple.TupleOutput;
import static calao.storage.DirectAccess.*;

public class TableDatabase
{

	final static DatabaseConfig DB_CONFIG = new DatabaseConfig();

	static {
		DB_CONFIG.setAllowCreate(true);
	}

	final BdbStorage storage;
	final int id;
	final Table table;
	final Index primaryIndex;
	final Database db;
	final IndexDatabase[] indexDBs;

	TableDatabase(BdbStorage storage, int id, Table table)
	throws DatabaseException
	{
		this.storage = storage;
		this.id = id;
		this.table = table;
		this.primaryIndex = table.getPrimaryKeyIndex();
		this.db = BdbStorage.env.openDatabase(null, 
				getDbName(storage, id), DB_CONFIG);
		indexDBs = new IndexDatabase[table.getIndexCount()-1];
		// open databases for secondary indexes
		int count = 0;
		for (Index index : table.getIndexList()) {
			if (index!=primaryIndex) {
				indexDBs[count++] = new IndexDatabase(this, index);
			}
		}
		// make sure all table fields have an id in the registry
		for (Field field : table.getFieldList()) {
			BdbStorage.envRegistry.getId(field);
		}
		log(DEBUG, "Table database %s opened", db.getDatabaseName());
	}

	static String getDbName(BdbStorage storage, int id)
	{
		return Integer.toString(storage.id)+"."+Integer.toString(id);
	}

	static void removeDatabase(BdbStorage storage, int id)
	throws DatabaseException
	{
		String dbName = getDbName(storage, id);
		BdbStorage.env.removeDatabase(null, dbName);
		log(DEBUG, "Database %s removed", dbName);
	}

	void close()
	throws DatabaseException
	{
		db.close();
	}

	void store(Record record)
	throws DatabaseException
	{
		RecordWriter writer = new RecordWriter(table);
		writer.write(record);
		db.put(null, writer.pkeyEntry, writer.dataEntry);
	}

	void replace(Record oldRec, Record newRec)
	throws DatabaseException
	{
		RecordWriter writer = new RecordWriter(table);
		writer.writePrimaryKey(oldRec);
		writer.writeData(newRec);
		db.put(null, writer.pkeyEntry, writer.dataEntry);
	}

	public Cursor<Record> listResults(final Search search) 
	throws DatabaseException
	{
		// build a list of usable cursors
		Vector<TableCursor> cursors = new Vector<TableCursor>();
		// try to match field conditions against indexes
		for (Index index : table.getIndexList()) {
			KeyWriter writer = new KeyWriter(this, index);
			FieldCondition[] keyConds = new FieldCondition[index.getFieldCount()];
			FieldCondition lastCond = null;
			RangeCondition rangeCond = null;
			for (int n=0; n<index.getFieldCount(); n++) {
				Field field = index.getField(n);
				EqualCondition equalCond = search.getFirstEqualCondition(field);
				if (equalCond!=null) {
					lastCond = keyConds[n] = equalCond;
					writer.setValueCount(field, 1);
					writer.writeValue(field, equalCond.getValue());
				}
				else {
					rangeCond = search.getFirstRangeCondition(field);
					if (rangeCond!=null) {
						lastCond = keyConds[n] = rangeCond;
						if (rangeCond.hasMinValue() && rangeCond.hasMaxValue()) {
							writer.setValueCount(field, 2);
							writer.writeValue(field, rangeCond.getMinValue());
							writer.writeValue(field, rangeCond.getMaxValue());
						}
						else if (rangeCond.hasMinValue()) {
							writer.setValueCount(field, 1);
							writer.writeValue(field, rangeCond.getMinValue());
						}
						else if (rangeCond.hasMaxValue()) {
							writer.setValueCount(field, 1);
							writer.writeValue(field, rangeCond.getMaxValue());
						}
					}
					break;
				}
			}
			if (lastCond!=null) {
				TableCursor cursor = null;
				if (writer.hasCompleteKeys() && rangeCond==null) {
					// equal conditions on all index fields
					SearchKeyCursor skCursor = new SearchKeyCursor(this);
					skCursor.setKey(Util.toEntry(writer.keys[0]));
					cursor = skCursor;		
				}
				else {
					KeyRangeCursor krCursor = new KeyRangeCursor(this);
					if (rangeCond==null) {
						DatabaseEntry key = Util.toEntry(writer.keys[0]);
						krCursor.setMinKey(key);
						krCursor.setMaxKey(key);						
					}
					else if (rangeCond.hasMinValue() && rangeCond.hasMaxValue()) {
						krCursor.setMinKey(Util.toEntry(writer.keys[0]));
						krCursor.setMaxKey(Util.toEntry(writer.keys[1]));
					}
					else if (rangeCond.hasMinValue()) {
						krCursor.setMaxKey(Util.toEntry(writer.keys[0]));
					}
					else if (rangeCond.hasMaxValue()) {
						krCursor.setMaxKey(Util.toEntry(writer.keys[0]));
					}
					cursor = krCursor;
				}
				if (cursor!=null) {
					cursor.use(index, writer.indexPos+1);
					for (FieldCondition cond : search.getConditions()) {
						if (!Util.arrayContains(keyConds, cond)) {
							cursor.addFetchCondition(cond);
						}
					}
					cursor.setDistinct(search.getDistinct());
					cursors.add(cursor);
				}
			}
		}
		// defaults to full-iteration cursor that checks all field conditions
		TableCursor cursor = !cursors.isEmpty()? findBestCursor(cursors) 
				: new TableCursor(this, search.getConditions()); 
		for (Field field : table.getFieldList()) {
			if (!search.isSearched(field)) cursor.skip(field);
		}
		if (search.getSortFields()!=null) {
			return new RecordSorter(cursor, search.getSortFields()).sort(); 
		}
		else return cursor;
	}

	TableCursor findBestCursor(Vector<TableCursor> cursors)
	{
		TableCursor best = null;
		for (TableCursor cursor : cursors) {
			if (best==null || best.indexUsedFieldCount<cursor.indexUsedFieldCount) {
				best = cursor;
			}
		}
		return best;
	}

	public Cursor<Record> list() 
	{
		return new TableCursor(this);
	}

	public Cursor<Record> list(long offset) 
	{
		TableCursor cursor = new TableCursor(this);
		cursor.skipNext(offset);
		return cursor;
	}

	Record fetch(Object[] keyValues)
	throws DatabaseException
	{
		DatabaseEntry data = new DatabaseEntry();
		OperationStatus status = db.get(null, makeKey(keyValues), data, LockMode.DEFAULT);
		if (status==OperationStatus.SUCCESS) {
			FetchReader reader = new FetchReader(table);
			Record record = reader.record;
			for (int i=0; i<primaryIndex.getFieldCount(); i++) {
				set(record, primaryIndex.getField(i), keyValues[i]);
			}
			reader.readData(data);
			return record;
		}
		else return null;
	}

	DatabaseEntry makeKey(Object[] keyValues) 
	throws DatabaseException
	{
		if (keyValues.length!=primaryIndex.getFieldCount()) {
			throw new IllegalArgumentException("Bad number of values");
		}
		Record record = new Record(table);
		int index = 0;
		TupleOutput key = new TupleOutput();
		for (Field field : primaryIndex) {
			Object value = keyValues[index++];
			RecordWriter.writeValue(field, value, key);
			set(record, field, value);
		}
		return Util.toEntry(key);
	}

	void drop(Object[] keyValues)
	throws DatabaseException
	{
		db.delete(null, makeKey(keyValues));
	}

	void clear()
	{
		TableCursor cursor = new TableCursor(this);
		try {
			while (cursor.next()) cursor.delete();
		}
		finally {
			cursor.close();
		}
	}

}
